package com.xxcloud.search.index.note.impl;

import java.util.ArrayList;
import java.util.List;
import javax.annotation.Resource;
import org.apache.commons.collections4.CollectionUtils;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.query.Criteria;
import org.springframework.data.elasticsearch.core.query.CriteriaQuery;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.data.elasticsearch.core.query.UpdateQuery;
import org.springframework.data.elasticsearch.core.query.UpdateQueryBuilder;
import org.springframework.stereotype.Service;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.collect.Lists;
import com.xxcloud.common.constants.Constants;
import com.xxcloud.model.web.JobResult;
import com.xxcloud.model.web.PageManager;
import com.xxcloud.search.etl.note.INoteEtlService;
import com.xxcloud.search.etl.note.bean.NoteEtlQuery;
import com.xxcloud.search.index.note.INoteIndexService;
import com.xxcloud.search.model.note.NoteField;
import com.xxcloud.search.model.note.NoteModel;

/**
 * Created by chenbingwen on 19/10/29.
 */
@Service("noteIndexService")
public class NoteIndexServiceImpl implements INoteIndexService {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    private ElasticsearchTemplate elasticsearchTemplate;
    @Resource(name = "noteEtlService")
    private INoteEtlService noteEtlService;

    @Override
    public JobResult rebuild() {
        JobResult jobResult = new JobResult();
        try {
            jobResult.begin();
            // 定义索引
            boolean ifCreateIndex = elasticsearchTemplate.createIndex(NoteModel.class);
            if (!ifCreateIndex) {
                jobResult.failed(Constants.INDEX_CREATE_ERROR);
            }
            int total = noteEtlService.count(new NoteEtlQuery());
            if (total > 0) {
                NoteEtlQuery query = null;
                PageManager pm = new PageManager(total, Constants.REBUILD_INDEX_BATCH_COUNT);
                while (pm.hasNextPage()) {
                    query = new NoteEtlQuery();
                    query.setPageNo(pm.getFirst());
                    query.setPageSize(pm.getMax());
                    List<IndexQuery> bulkIndexQueries = new ArrayList<>();
                    List<NoteModel> noteModels = noteEtlService.getNoteModels(query);
                    if (CollectionUtils.isNotEmpty(noteModels)) {
                        noteModels.stream().forEach(noteModel -> {
                            IndexQuery indexQuery = new IndexQueryBuilder().withId(String.valueOf(noteModel.getId())).withObject(noteModel).build();
                            bulkIndexQueries.add(indexQuery);
                        });
                        elasticsearchTemplate.bulkIndex(bulkIndexQueries);
                    }
                }
            }
        } catch (Exception ex) {
            logger.error("noteIndexService rebuild error:{}", ex);
        } finally {
            jobResult.finish();
        }
        return jobResult;
    }

    @Override
    public JobResult updateByNoteIds(List<Long> noteIds) {
        JobResult jobResult = new JobResult();
        if (CollectionUtils.isEmpty(noteIds)) {
            return jobResult.failed(Constants.PARAM_ERROR);
        }
        try {
            jobResult.begin();
            List<List<Long>> subNoteIds = Lists.partition(noteIds, Constants.UPDATE_INDEX_BATCH_COUNT);
            for (List<Long> tmpSubNoteIds : subNoteIds) {
                // 删除老索引数据
                deleteByNoteIds(tmpSubNoteIds);
                //
                NoteEtlQuery query = new NoteEtlQuery();
                query.setNoteIds(tmpSubNoteIds);
                List<NoteModel> noteModels = noteEtlService.getNoteModels(query);

                List<UpdateQuery> updateQueries = new ArrayList<>();

                noteModels.stream().forEach(noteModel -> {
                    UpdateQuery updateQuery = new UpdateQueryBuilder().withId(String.valueOf(noteModel.getId())).withClass(NoteModel.class)
                            .withUpdateRequest(
                                    new UpdateRequest().doc(JSONObject.toJSONString(noteModel, SerializerFeature.WriteMapNullValue), XContentType.JSON))
                            .build();
                    if (updateQuery != null) {
                        updateQueries.add(updateQuery);
                    }
                });
                elasticsearchTemplate.bulkUpdate(updateQueries);
            }
        } catch (Exception ex) {
            logger.error("noteIndexService updateByNoteIds error:{}", ex);
        } finally {
            jobResult.finish();
        }
        return jobResult;
    }

    @Override
    public JobResult deleteByNoteIds(List<Long> noteIds) {
        JobResult jobResult = new JobResult();
        try {
            jobResult.begin();
            CriteriaQuery deleteQuery = new CriteriaQuery(new Criteria().and(NoteField.NOTE_ID).in(noteIds));
            elasticsearchTemplate.delete(deleteQuery, getIndexClass());
        } catch (Exception ex) {
            logger.error("noteIndexService deleteByNoteIds error:{}", ex);
        } finally {
            jobResult.finish();
        }
        return jobResult;
    }

    @Override
    public void deleteIndex() {
        elasticsearchTemplate.deleteIndex(getIndexClass());
    }

    private Class<NoteModel> getIndexClass() {
        return NoteModel.class;
    }
}
